package cn.doitedu.api;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.Map;

public class _12_MapState_ListState_Api {

    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<String> stream = env.socketTextStream("doitedu", 9999);


        stream.keyBy(new KeySelector<String, String>() {
                    @Override
                    public String getKey(String value) throws Exception {
                        return value.split(",")[0];
                    }
                })
                .map(new RichMapFunction<String, String>() {


                    MapState<String, String> mapState;
                    ListState<String> listState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        RuntimeContext runtimeContext = getRuntimeContext();

                        // 获取mapState
                        this.mapState = runtimeContext.getMapState(new MapStateDescriptor<String, String>("map-state", String.class, String.class));

                        // 获取listState
                        this.listState = runtimeContext.getListState(new ListStateDescriptor<String>("list-state", String.class));

                    }

                    @Override
                    public String map(String value) throws Exception {

                        /**
                         * mapState的数据操作api
                         */
                        // 往 mapState  添加数据
                        mapState.put("a","zhangSan");
                        mapState.put("b","liSi");
                        mapState.put("a","jack"); // 如果mapState已存在的key，被put，则会覆盖值


                        // 从mapState 取数据
                        String s = mapState.get("a");

                        // 判断mapState中是否存在指定的key
                        if(mapState.contains("a")){
                            // TODO
                        }

                        // 迭代遍历 mapState
                        Iterable<Map.Entry<String, String>> entries = mapState.entries();
                        for (Map.Entry<String, String> entry : entries) {
                            String ruleId = entry.getKey();
                            String value1 = entry.getValue();
                        }


                        /**
                         * listState的数据操作api
                         */
                        // 添加元素
                        listState.add("a");
                        listState.addAll(Arrays.asList("b","c","d"));

                        // 取数据，不能像ArrayList一样一次取指定index上的数据，而是只能取出一个迭代器，进行遍历
                        Iterable<String> strings = listState.get();
                        for (String str : strings) {

                        }


                        return null;
                    }
                });


    }
}
